package io.dy.custom_pool;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author : dmj
 * @date: 2024/7/11 10:42 上午
 * @description: 自定义阻塞队列
 */
public class CustomBlockQueue<T> {

    /**
     * 任务队列
     */
    private final Deque<T> queue = new ArrayDeque<>();

    /**
     * 锁
     */
    private final ReentrantLock lock = new ReentrantLock();

    /**
     * 生产者条件变量 标识队列已经满了 无需生产 进入等待
     */
    private final Condition fullWaitSet = lock.newCondition();

    /**
     * 消费者条件变量 标识队列已经空了 无法消费 进入等待
     */
    private final Condition emptyWaitSet = lock.newCondition();

    /**
     * 队列容量
     */
    private final int capacity;


    public CustomBlockQueue(int capacity) {
        this.capacity = capacity;
    }

    /**
     * 阻塞获取元素
     * @return
     */
    public T take(){
        //1、加锁
        lock.lock();
        try {
            //2、判断队列是否为空 为空 进入等待
            while (queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //3、执行到这 标识 队列不为空 可以获取原属
            T t = queue.removeFirst();

            //4、获取到元素后 需要唤醒 fullWaitSet 中的生产线程开始工作
            fullWaitSet.signal();

            return t;

        }finally {
            lock.unlock();
        }
    }

    /**
     * 阻塞获取元素 带超时时间
     * @return
     */
    public T takeWithTimeout(long time, TimeUnit unit){
        //1、加锁
        lock.lock();
        try {
            //2、转换等待时间为 纳秒
            long nanos = unit.toNanos(time);
            //3、判断队列是否为空 为空 进入等待
            while (queue.isEmpty()){
                try {
                    //3.1、判断等待时间是否小于0 小于则直接返回null
                    if(nanos < 0){return null;}
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //4、执行到这 标识 队列不为空 可以获取元素
            T t = queue.removeFirst();

            //5、获取到元素后 需要唤醒 fullWaitSet 中的生产线程开工作
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    /**
     * 阻塞添加元素
     * @param element
     */
    public void add(T element){
        //1、加锁
        lock.lock();
        try{
            //2、判断队列大小是否已经达到容量上线
            while (queue.size() == capacity){
                try {
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //3、执行到这标识容量未到达上线 加入元素
            queue.addLast(element);

            //4、唤醒 emptyWaitSet 获取元素
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }

    }

    /**
     * 阻塞添加元素 带超时时间
     * @param task
     * @param timeout
     * @param unit
     */
    public boolean add(T task, long timeout, TimeUnit unit){
        //1、加锁
        lock.lock();
        try{
            long nanos = unit.toNanos(timeout);
            //2、判断队列大小是否已经达到容量上线
            while (queue.size() == capacity){
                try {
                    if(nanos < 0){return false;}
                    nanos = fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //3、执行到这标识容量未到达上线 加入元素
            queue.addLast(task);

            //4、唤醒 emptyWaitSet 获取元素
            emptyWaitSet.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }

    /**
     * 添加元素 带拒绝策略
     * @param policy
     * @param task
     */
    public void addWithRejectPolicy(CustomRejectPolicy<T> policy, T task){
        //1、加锁
        lock.lock();
        try{
            //2、判断队列是否已满
            if(queue.size() == capacity){
                //交与拒绝策略来实现具体的逻辑
                policy.reject(this, task);
            }else{
                //队列未满 直接加入队列
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        }finally {
            lock.unlock();
        }
    }

    /**
     * 获取队列大小
     * @return
     */
    public int size(){
        lock.lock();
        try {
            return queue.size();
        }finally {
            lock.unlock();
        }
    }
}
